package streaming.api.tableapi;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Csv;
import org.apache.flink.table.descriptors.FileSystem;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.types.Row;

/**
 * 从文件获取数据
 * 简单查询
 * 输出到文件
 */
public class TableTest4_file1 {

    public static void main(String[] args) throws Exception {
        // 1. 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        // 2. 表的创建：连接外部系统，读取数据
        String fromPath = "D:\\IdeaProjects\\springboot-flink-1\\flinkTutorial\\src\\main\\resources\\sensor.txt";
        String toPath = "D:\\IdeaProjects\\springboot-flink-1\\flinkTutorial\\src\\main\\resources\\sensor2.txt";
        // 定义表结构
        Schema schema1 = new Schema()
                .field("id", DataTypes.STRING())
                .field("timestamp", DataTypes.BIGINT())
                .field("temp", DataTypes.DOUBLE());
        tableEnv.connect(new FileSystem().path(fromPath))
                .withFormat(new Csv())
                .withSchema(schema1)
                .createTemporaryTable("inputTable");

        // 简单查询
        Table inputTable = tableEnv.from("inputTable");
        Table resultTable = inputTable.select("id, temp").filter("id === 'sensor_c'");
        // 打印输出
        tableEnv.toAppendStream(resultTable, Row.class).print("resultTable:");

        // 4. 输出到文件  连接外部文件注册输出表
        Schema schema2 = new Schema()
                .field("id", DataTypes.STRING())
                .field("temp", DataTypes.DOUBLE());
        tableEnv.connect(new FileSystem().path(toPath))
                .withFormat(new Csv())
                .withSchema(schema2)
                .createTemporaryTable("outputTable");
        resultTable.executeInsert("outputTable");
    }

}
